package com.heima.kafka.listener;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;
import java.util.Arrays;

/**
 * @BelongsProject: heima-leadnews
 * @BelongsPackage: com.heima.kafka.listener
 * @Author: wangjian
 * @CreateTime: 2022-07-08  17:36
 * @Description: TODO
 * @Version: 1.0
 */
public class WordCountListener {
    @StreamListener("WordCountSource")
    @SendTo("WordCountResult")
    public KStream<String, String> process(KStream<String, String> input) {


        return input.flatMapValues((ValueMapper<String, Iterable<String>>) value -> { //value  就是原数据
            return Arrays.asList(value.split(" "));
        }).groupBy((key, value) -> {
            return value; //根据value分组
        }).windowedBy(TimeWindows.of(Duration.ofSeconds(60)))
                .count(Materialized.with(Serdes.String(), Serdes.Long()))
                .mapValues(value -> value.toString()).toStream().map( (key, value) -> new KeyValue<>(key.key(), value));

    }

}
